package com.wjw.redis.stream.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.wjw.redis.stream.model.BaseMsgEvent;
import com.wjw.redis.stream.model.MessageEvent;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RFuture;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

/**
 * @author wjwei
 */
@Slf4j
@Component
public class EventPublisher implements ApplicationContextAware {
    private static RedissonClient redissonClient;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        redissonClient = applicationContext.getBean(RedissonClient.class);
    }

    public static void publish(BaseMsgEvent baseMsgEvent) {
        if (baseMsgEvent == null) {
            return;
        }
        if (baseMsgEvent instanceof MessageEvent) {
            RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque("redis-block-test");
//            blockingDeque.addLast(((MessageEvent) baseMsgEvent).getMessage());
            RFuture<Void> future = blockingDeque.putAsync(JSON.toJSONString(baseMsgEvent, SerializerFeature.WriteClassName));
            future.whenComplete((s, t) -> {
                if (t != null) {
                    log.error("发送异常：", t);
                } else {
                    log.info("消息发送成功");
                }
            });
            log.info("发消息：{}", JSON.toJSONString(baseMsgEvent));
        }
    }

    public static void publish(String channel, Object message) {
        redissonClient.getTopic(channel).publish(message);
    }
}
